package com.xiehua.task.client.handler;

import com.xiehua.task.client.job.JobExecuter;
import com.xiehua.task.intf.AbsBusinessHandler;
import com.xiehua.task.job.pool.DelayJob;
import com.xiehua.task.job.pool.FixedJob;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.msg.TaskExecuteCompleteReqBody;
import com.xiehua.task.msg.TaskPushReqBody;
import com.xiehua.task.msg.TaskPushRespBody;
import com.xiehua.task.prop.XiehuaConfig;
import com.xiehua.task.protocol.Type;
import com.xiehua.task.protocol.TaskPacket;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.Aio;
import org.tio.core.ChannelContext;
import org.tio.utils.json.Json;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

import static com.xiehua.task.job.pool.Job.JobType.delay;
import static com.xiehua.task.job.pool.Job.JobType.fixed;
import static com.xiehua.task.msg.BaseBody.RESULT_FAIL;
import static com.xiehua.task.msg.BaseBody.RESULT_SUCCESS;
import static com.xiehua.task.prop.XiehuaConfig.KEY_CLIENT_DELAY_TASK_EXECUTER;
import static com.xiehua.task.prop.XiehuaConfig.KEY_CLIENT_SECRET_KEY;

/**
 * @author Mr.zhang
 * 2018年3月27日 下午9:51:28
 */
public class TaskPushReqHandler extends AbsBusinessHandler<TaskPushReqBody> {

	private static Logger log = LoggerFactory.getLogger(TaskPushReqHandler.class);

	// 创建容量为NUMBER的线程池。
	public static ExecutorService pool;

	static {
		pool = Executors.newFixedThreadPool(10);
	}


	AtomicLong tokenSeq = new AtomicLong();

	/**
	 *
	 * @author tanyaowu
	 */
	public TaskPushReqHandler() {

	}

	/**
	 * @return
	 * @author tanyaowu
	 */
	@Override
	public Class<TaskPushReqBody> bodyClass() {
		return TaskPushReqBody.class;
	}

	/**
	 * @param packet
	 * @param bsBody
	 * @param channelContext
	 * @return
	 * @throws Exception
	 * @author tanyaowu
	 */
	@Override
	public Object handler(TaskPacket packet, TaskPushReqBody bsBody, ChannelContext channelContext) throws Exception {
		log.info("收到来自服务端推送的task消息执行命令:{}", Json.toJson(bsBody));
		//build TaskPushRespBody
		TaskPushRespBody taskPushRespBody = new TaskPushRespBody();
		taskPushRespBody.setResult(RESULT_SUCCESS);
		taskPushRespBody.setJobId(bsBody.getJobId());
		//notify
		notify(Type.TASK_PUSH_RESP,channelContext,packet.getClientId(),taskPushRespBody);
		//parse job
		if(delay.equals(bsBody.getType())){
			DelayJob job = Json.toBean(bsBody.getJobContent(),DelayJob.class);
			executerTask(job, channelContext ,packet.getClientId());
		}
		if(fixed.equals(bsBody.getType())){
			FixedJob job =  Json.toBean(bsBody.getJobContent(),FixedJob.class);
			executerTask(job, channelContext ,packet.getClientId());
		}
		return null;
	}

	/**
	 * 执行任务
	 * **/
	private void executerTask(final Job job,ChannelContext channelContext,String cid) throws InterruptedException, ExecutionException ,UnsupportedEncodingException{
		Map<String,String> configMap = XiehuaConfig.getInstance();
		if(job.getType().equals(delay) && StringUtils.isEmpty(configMap.get(KEY_CLIENT_DELAY_TASK_EXECUTER))){
			log.error("延迟任务执行器不存在:{}",KEY_CLIENT_DELAY_TASK_EXECUTER);
			return ;
		}
		if(job.getType().equals(delay)){
			DelayJob delayJob = (DelayJob) job;
			FutureTask<Boolean> futureTask = new FutureTask<>(()->{
				try {
					Class jobExecuterClass = Class.forName(configMap.get(KEY_CLIENT_DELAY_TASK_EXECUTER));
					JobExecuter executer = (JobExecuter) jobExecuterClass.newInstance();
					return executer.asynExecute(delayJob.getBody());
				}catch (ClassNotFoundException e){
					log.error("客户端配置的executer不能实例化",e);
					return false;
				}
			});
			pool.submit(futureTask);
			//build req body
			TaskExecuteCompleteReqBody reqBody = new TaskExecuteCompleteReqBody();
			reqBody.setJobId(job.getJobId());
			if(futureTask.get()){//通知服务器执行成功
				reqBody.setResult(RESULT_SUCCESS);
			}else{
				reqBody.setResult(RESULT_FAIL);
			}
			//执行失败
			notify(Type.TASK_EXECUTE_COMPLETE_REQ,channelContext,cid,reqBody);
			return ;
		}
		if(job.getType().equals(fixed)){
			FixedJob fixedJob = (FixedJob) job;
			FutureTask<Boolean> futureTask = new FutureTask<>(()->{
				try {
					Class jobExecuterClass = Class.forName(fixedJob.getClassName());
					Method method = jobExecuterClass.getMethod(fixedJob.getMethodName());
					method.invoke(jobExecuterClass.newInstance());
					return true;
				}catch (ClassNotFoundException e){
					log.error("客户端配置的executer不能实例化",e);
					return false;
				}
			});
			pool.submit(futureTask);
			//build req body
			TaskExecuteCompleteReqBody reqBody = new TaskExecuteCompleteReqBody();
			reqBody.setJobId(job.getJobId());
			if(futureTask.get()){//通知服务器执行成功
				reqBody.setResult(RESULT_SUCCESS);
			}else{
				reqBody.setResult(RESULT_FAIL);
			}
			//执行失败
			notify(Type.TASK_EXECUTE_COMPLETE_REQ,channelContext,cid,reqBody);
			return ;
		}

	}

	/**
	 * notify server
	 * */
	private void notify(byte type, ChannelContext channelContext,String cid,Object bizOb) throws UnsupportedEncodingException {
		//resp packet
		TaskPacket respPacket = new TaskPacket(type,cid);
		String bizData = Json.toJson(bizOb);
		respPacket.setBody(bizData.getBytes(TaskPacket.CHARSET));
		respPacket.signData(bizData, XiehuaConfig.getInstance().get(KEY_CLIENT_SECRET_KEY));
		Aio.send(channelContext,respPacket);
	}

	private String newToken() {
		return System.currentTimeMillis() + "_" + tokenSeq.incrementAndGet();
	}

}
